package com.atguigu.udmp.stream.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.udmp.stream.bean.UserEvent;
import com.atguigu.udmp.stream.bean.UserEventDefine;
import com.atguigu.udmp.stream.common.util.GroovyUtil;
import com.atguigu.udmp.stream.extractor.UserEventExtractor;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Map;

public class UserEventExtractProcessFunction extends BroadcastProcessFunction<JSONObject,JSONObject, UserEvent> {

    MapStateDescriptor<Long,UserEventDefine> userEventDefineMapStateDescriptor;


    public UserEventExtractProcessFunction(MapStateDescriptor<Long, UserEventDefine> userEventDefineMapStateDescriptor){
        this.userEventDefineMapStateDescriptor=userEventDefineMapStateDescriptor;
    }

    /*
     根据广播状态中的事件定义，从数据对象中提取用户事件
     */
    @Override
    public void processElement(JSONObject dataJsonObj, BroadcastProcessFunction<JSONObject, JSONObject, UserEvent>.ReadOnlyContext ctx, Collector<UserEvent> out) throws Exception {
        ReadOnlyBroadcastState<Long, UserEventDefine> broadcastState = ctx.getBroadcastState(userEventDefineMapStateDescriptor);

        //遍历有的事件定义，找到对应匹配
        for (Map.Entry<Long, UserEventDefine> userEventDefineEntry : broadcastState.immutableEntries()) {
            UserEventDefine userEventDefine = userEventDefineEntry.getValue();
            UserEventExtractor userEventExtractor = userEventDefine.getUserEventExtractor();

            UserEvent userEvent = userEventExtractor.extract(dataJsonObj,userEventDefine);

            if(userEvent!=null){
                out.collect(userEvent);
            }

        }
    }

    @Override
    public void processBroadcastElement(JSONObject userEventDefineJsonObj, BroadcastProcessFunction<JSONObject, JSONObject, UserEvent>.Context ctx, Collector<UserEvent> out) throws Exception {
        //4.1 加载用户事件定义到广播状态
        // 获得定义信息
        BroadcastState<Long, UserEventDefine> broadcastState = ctx.getBroadcastState(userEventDefineMapStateDescriptor);

        // 获得操作类型
        String opType=userEventDefineJsonObj.getString("op");
        JSONObject dataAfterJsonObj = userEventDefineJsonObj.getJSONObject("after");
        UserEventDefine userEventDefine = dataAfterJsonObj.toJavaObject(UserEventDefine.class);
        // 处理c 创建 , r 预加载, u 修改
        if(opType.equals("c")||opType.equals("r")||opType.equals("u")){
            // 从jsonObj中提取数据 加载到userEventDefine
            try{
                // 通过Groovy工具加载为对象，并赋值给 userEventDefine
                UserEventExtractor userEventExtractor = GroovyUtil.parseGroovyToObject(userEventDefine.getEventGroovyCode(), UserEventExtractor.class);
                userEventDefine.setUserEventExtractor(userEventExtractor);

                broadcastState.put(userEventDefine.getId(),userEventDefine);
            }catch (Exception e){
                throw  new RuntimeException("事件处理器生成有误");
            }

        } else if (opType.equals("d")) {
            broadcastState.remove(userEventDefine.getId() );
        }

    }
}
